package com.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
 * @author : ranzlupup
 * @date : 2023/3/2 20:06
 */
object SparkSQL_Basic {
    def main(args: Array[String]): Unit = {

        // TODO 创建SparkSQL的运行环境
        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
        val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

        // TODO 引入隐式转换
        import spark.implicits._

        // TODO 执行逻辑操作
        // val df: DataFrame = spark.read.json("datas/user.json")
        // df.show()

        // df.createOrReplaceTempView("user")
        // spark.sql("select * from user").show()
        // spark.sql("select age from user").show()
        // spark.sql("select name from user").show()
        // spark.sql("select avg(age) from user").show()

        // df.select($"age" + 1).show()
        // df.select('age + 1).show()

        // RDD <=> DataFrame
        val rdd: RDD[(String, String, Int)] = spark.sparkContext.makeRDD(List(("1", "rzl", 22), ("2", "zhb", 22)))
        val df: DataFrame = rdd.toDF("id", "name", "age")
        // df.show
        val rdd1: RDD[Row] = df.rdd
        // rdd1.collect().foreach(println)

        // DataFrame <=> DataSet
        val ds: Dataset[User] = df.as[User]
        // ds.show()
        val df2: DataFrame = ds.toDF()
        // df2.show()

        // RDD <=>DataSet
        val ds2: Dataset[User] = rdd.map {
            case (id, name, age) => {
                User(id, name, age)
            }
        }.toDS()

        ds2.show()
        val rdd2: RDD[Row] = df2.rdd
        rdd2.collect().foreach(
            item => {
                println(item.getString(0))
                println(item.getString(1))
            }
        )
        println(rdd2.count())


        // val dataFrame: DataFrame = spark.read.json("datas/user.json")
        //
        // // SQL风格语法
        // dataFrame.createOrReplaceTempView("user")
        // spark.sql("select * from user").show()
        //
        // // DSL风格语法
        // dataFrame.select("name", "age").show()

        // TODO 关闭环境
        spark.stop()
    }
}

case class User(id: String, name: String, age: Int)

